package com.auxgroup.smarthome.mqtt;

import com.google.common.util.concurrent.ThreadFactoryBuilder;
import org.fusesource.hawtbuf.Buffer;
import org.fusesource.hawtbuf.UTF8Buffer;
import org.fusesource.mqtt.client.Callback;
import org.fusesource.mqtt.client.Future;
import org.fusesource.mqtt.client.FutureConnection;
import org.fusesource.mqtt.client.MQTT;
import org.fusesource.mqtt.client.MQTTException;
import org.fusesource.mqtt.client.Message;
import org.fusesource.mqtt.client.QoS;
import org.fusesource.mqtt.client.Topic;
import org.fusesource.mqtt.codec.CONNACK;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.net.URISyntaxException;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.UUID;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * MQTT 客户端结合业务实现
 *
 * @author Qiuhua Lai
 * @email qhlai@gizwits
 */
public class AuxMqttClient {
    private static final Logger log = LoggerFactory.getLogger(AuxMqttClient.class);
    private MQTT mqtt;
    private FutureConnection connection;
    private volatile List<Topic> topics = new ArrayList<>();
    private ExecutorService executorService = Executors.newScheduledThreadPool(12, new ThreadFactoryBuilder().setNameFormat("Aux-Mqtt").build());

    public AuxMqttClient(MqttConfProp mqttConfProp) {
        try {
            this.mqtt = new MQTT();
            this.mqtt.setClientId(UUID.randomUUID().toString());
            this.mqtt.setHost(mqttConfProp.getHost(), mqttConfProp.getPort());
            this.mqtt.setCleanSession(mqttConfProp.isCleanSession());
            this.mqtt.setKeepAlive(mqttConfProp.getKeepAlive());
            this.mqtt.setReconnectAttemptsMax(mqttConfProp.getReconnectAttemptsMax());
            this.mqtt.setReconnectDelay(mqttConfProp.getReconnectDalay());
            this.mqtt.setSendBufferSize(mqttConfProp.getSendBufferSize());
            this.mqtt.setWillQos(mqttConfProp.getWillQoS());
            this.mqtt.setWillTopic(mqttConfProp.getWillTopic());
            this.mqtt.setWillMessage(mqttConfProp.getWillMessage());
            this.mqtt.setVersion(mqttConfProp.getVersion());
        } catch (URISyntaxException e) {
            e.printStackTrace();
        }
    }

    /**
     * 建立MQTT 连接， 等待2S，若未建立连接则跳过，无阻塞等待重连
     */
    public void connect() {
        connection = mqtt.futureConnection();
        Future<Void> future = connection.connect();
        try {
            future.await();
        } catch (Exception e) {
            log.error("MQTT Connect Timeout!", e);
        }
        future.then(new Callback<Void>() {
            @Override
            public void onSuccess(Void value) {
                try {
                    subscribe();
                } catch (MQTTException e) {
                    log.debug("Mqtt subscribe Exception", e);
                }
                log.info("MQTT connect success!");
            }

            @Override
            public void onFailure(Throwable value) {
                if (log.isDebugEnabled()) {
                    value.printStackTrace();
                }
                log.error("MQTT connect fail", value);
            }
        });

    }

    /**
     * 开始订阅主题
     *
     * @return
     */
    public Future<byte[]> subscribe() throws MQTTException {
        if (!connection.isConnected()) {
            throw new MQTTException("MQTT unconnected", new CONNACK().code(CONNACK.Code.CONNECTION_REFUSED_SERVER_UNAVAILABLE));
        }
        if (log.isDebugEnabled()) {
            log.debug("订阅主题成功：{}", Arrays.toString(topics.toArray()));
        }
        synchronized (this) {
            try {
                Future<byte[]> future = connection.subscribe(topics.toArray(new Topic[topics.size()]));
                future.await();
                return future;
            } catch (Exception e) {
                log.debug("订阅主题失败", e);
            }
        }
        return null;
    }

    /**
     * 接收订阅的消息，自定义业务可通过回调方法实现
     *
     * @param callback
     */
    public void receive(Callback<Message> callback) {
        executorService.execute(() -> {
            try {
                while (connection.isConnected()) {
                    Future<Message> receive = connection.receive();
                    Message message = receive.await();
                    callback.onSuccess(message);
                    //签收消息的回执
                    message.ack();
                }
            } catch (Exception e) {
                callback.onFailure(e);
            }
        });
    }

    /**
     * 发布消息
     *
     * @param topic
     * @param payload
     * @param qos
     * @param retain
     * @return
     */
    public Future<Void> publish(final String topic, final byte[] payload, final QoS qos, final boolean retain) throws MQTTException {
        if (!connection.isConnected()) {
            throw new MQTTException("MQTT unconnected", new CONNACK().code(CONNACK.Code.CONNECTION_REFUSED_SERVER_UNAVAILABLE));
        }
        if (log.isDebugEnabled()) {
            log.debug("publish topic：clientId:{}, topicName:{}, payload:{}, qos:{}, retain:{}", mqtt.getClientId(), topic, new String(payload, StandardCharsets.UTF_8), qos, retain);
        }
        return connection.publish(topic, payload, qos, retain);
    }

    /**
     * 发布消息
     *
     * @param topic
     * @param payload
     * @param qos
     * @param retain
     * @return
     */
    public Future<Void> publish(final UTF8Buffer topic, final Buffer payload, final QoS qos, final boolean retain) throws MQTTException {
        if (!connection.isConnected()) {
            throw new MQTTException("MQTT unconnected", new CONNACK().code(CONNACK.Code.CONNECTION_REFUSED_SERVER_UNAVAILABLE));
        }
        if (log.isDebugEnabled()) {
            log.debug("publish topic：clientId:{}, topicName:{}, payload:{}, qos:{}, retain:{}", mqtt.getClientId(), topic, new String(payload.data, StandardCharsets.UTF_8), qos, retain);
        }
        return connection.publish(topic, payload, qos, retain);
    }

    public List<Topic> getTopics() {
        return topics;
    }

    public void setTopics(List<Topic> topics) {
        this.topics = topics;
    }

}
